package com.piece.core.redis.util;

import com.piece.core.framework.support.convert.Convert;
import com.piece.core.framework.util.lock.ZLock;
import com.piece.core.framework.util.string.JsonUtil;
import com.piece.core.framework.util.cache.Cache;
import com.piece.core.redis.lock.RedissonDistributedLock;
import lombok.extern.slf4j.Slf4j;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisClusterNode;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.RedisServerCommands;
import org.springframework.data.redis.core.*;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.SerializationUtils;
import org.springframework.util.Assert;
import javax.annotation.Resource;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;

/**
 * redis 基本操作,使用分布式锁操作
 */
@Slf4j
public class RedisCache implements Cache {

    @Resource
    private RedisTemplate redisTemplate;

    @Resource
    private ValueOperations<String, String> valueOperations;

    @Resource
    private HashOperations<String, String, Object> hashOperations;

    @Resource
    private ListOperations<String, Object> listOperations;

    @Resource
    private SetOperations<String, Object> setOperations;

    @Resource
    private ZSetOperations<String, Object> zSetOperations;

    @Resource
    private RedissonDistributedLock redissonDistributedLock;

    /**
     * 不设置过期时长
     */
    public final static long NOT_EXPIRE = -1;

    /**
     * 获取链接工厂
     */
    private RedisConnectionFactory getConnectionFactory() {
        return this.redisTemplate.getConnectionFactory();
    }

    /**
     * 获取 RedisTemplate对象
     */
    private RedisTemplate getRedisTemplate() {
        return redisTemplate;
    }

    @Override
    public boolean isPrimary() {
        return true;
    }

    /**
     * 查询所有key
     *
     * @param keyPatten redis主键patten
     */
    @Override
    public Set<String> keys(final String keyPatten) {
        return redisTemplate.keys(keyPatten + "*");
    }

    /**
     * 根据key获取对象
     *
     * @param key redis主键
     */
    @Override
    public byte[] get(final byte[] key) {
        return (byte[]) redisTemplate.execute((RedisCallback<byte[]>) connection -> connection.get(key));
    }

    /**
     * 根据key获取对象
     *
     * @param key redis主键
     */
    @Override
    public String get(String key) {
        return get(key, String.class, NOT_EXPIRE);
    }

    /**
     * 根据key获取对象
     *
     * @param key   redis主键
     * @param clazz 实例化对象
     */
    @Override
    public <T> T get(String key, Class<T> clazz) {
        return get(key, clazz, NOT_EXPIRE);
    }

    /**
     * 根据key获取对象并延长过期时间
     *
     * @param key   redis主键
     * @param clazz 实例化对象
     * @param time  过期时间(单位秒)
     */
    @Override
    public <T> T get(String key, Class<T> clazz, long time) {
        return get(key, clazz, time, TimeUnit.SECONDS);
    }

    /**
     * 根据key获取对象并延长过期时间
     *
     * @param key      redis主键
     * @param clazz    实例化对象
     * @param time     过期时间
     * @param timeUnit 过期时间单位
     */
    @Override
    public <T> T get(String key, Class<T> clazz, long time, TimeUnit timeUnit) {
        try {
            String value = valueOperations.get(key);
            if (NOT_EXPIRE != time) {
                redisTemplate.expire(key, time, timeUnit);
            }
            return (null == value) ? null : fromJson(value, clazz);
        } catch (Exception e) {
            return null;
        }
    }

    /**
     * 根据key获取对象
     *
     * @param key             the key
     * @param valueSerializer 序列化
     */
    public Object get(final String key, RedisSerializer<Object> valueSerializer) {
        byte[] rawKey = rawKey(key);
        return redisTemplate.execute(connection -> deserializeValue(connection.get(rawKey), valueSerializer), true);
    }

    /**
     * 添加非过期缓存
     *
     * @param key   redis主键
     * @param value 值
     */
    @Override
    public void set(final String key, final Object value) {
        ZLock zLock = redissonDistributedLock.lock(key, true);
        try {
            valueOperations.set(key, toJson(value));
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            redissonDistributedLock.unlock(zLock);
        }
    }

    /**
     * 一次性添加数组到非过期缓存
     *
     * @param keys   redis主键数组
     * @param values 值数组
     */
    @Override
    public void set(final String[] keys, final Object[] values) {
        ZLock zLock = redissonDistributedLock.lock(Convert.toString(keys), true);
        try {
            for (int i = 0; i < keys.length; i++) {
                valueOperations.set(keys[i], toJson(values[i]));
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            redissonDistributedLock.unlock(zLock);
        }
    }

    /**
     * 添加到带有[过期时间的]缓存
     *
     * @param key   redis主键
     * @param value 值
     * @param time  过期时间(单位秒)
     */
    @Override
    public void setExpire(final byte[] key, final byte[] value, final long time) {
        redisTemplate.execute((RedisCallback<Long>) connection -> {
            connection.set(key, value);
            if (NOT_EXPIRE != time) {
                connection.expire(key, time);
            }
            return 1L;
        });
    }

    /**
     * 添加到带有[过期时间的]缓存，时间单位：秒
     *
     * @param key   redis主键
     * @param value 值
     * @param time  过期时间
     */
    @Override
    public void setExpire(final String key, final Object value, final long time) {
        this.setExpire(key, value, time, TimeUnit.SECONDS);
    }

    /**
     * 添加到带有[过期时间的]缓存
     *
     * @param key      redis主键
     * @param value    值
     * @param time     过期时间
     * @param timeUnit 过期时间单位
     */
    @Override
    public void setExpire(final String key, final Object value, final long time, final TimeUnit timeUnit) {
        ZLock zLock = redissonDistributedLock.lock(key, true);
        try {
            valueOperations.set(key, toJson(value));
            if (NOT_EXPIRE != time) {
                redisTemplate.expire(key, time, timeUnit);
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            redissonDistributedLock.unlock(zLock);
        }
    }

    /**
     * 一次性添加数组到[过期时间的]缓存
     *
     * @param keys   redis主键数组
     * @param values 值数组
     * @param time   过期时间(单位秒)
     */
    @Override
    public void setExpire(final String[] keys, final Object[] values, final long time) {
        ZLock zLock = redissonDistributedLock.lock(Convert.toString(keys), true);
        try {
            for (int i = 0; i < keys.length; i++) {
                valueOperations.set(keys[i], toJson(values[i]), time, TimeUnit.SECONDS);
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            redissonDistributedLock.unlock(zLock);
        }
    }

    /**
     * 添加到带有[过期时间的]缓存
     *
     * @param key             redis主键
     * @param value           值
     * @param time            过期时间
     * @param timeUnit        过期时间单位
     * @param valueSerializer 自定义值序列化方法
     */
    public void setExpire(final String key, final Object value, final long time, final TimeUnit timeUnit, RedisSerializer<Object> valueSerializer) {
        byte[] rawKey = rawKey(key);
        byte[] rawValue = rawValue(value, valueSerializer);

        redisTemplate.execute(new RedisCallback<Object>() {
            @Override
            public Object doInRedis(RedisConnection connection) throws DataAccessException {
                potentiallyUsePsetEx(connection);
                return null;
            }

            public void potentiallyUsePsetEx(RedisConnection connection) {
                if (!TimeUnit.MILLISECONDS.equals(timeUnit) || !failsafeInvokePsetEx(connection)) {
                    connection.setEx(rawKey, TimeoutUtils.toSeconds(time, timeUnit), rawValue);
                }
            }

            private boolean failsafeInvokePsetEx(RedisConnection connection) {
                boolean failed = false;
                try {
                    connection.pSetEx(rawKey, time, rawValue);
                } catch (UnsupportedOperationException e) {
                    failed = true;
                }
                return !failed;
            }
        }, true);
    }

    /**
     * 设置缓存过期时间
     *
     * @param key  the keys
     * @param time 过期时间(单位秒)
     */
    @Override
    public void setExpire(String key, long time) {
        redisTemplate.expire(key, time, TimeUnit.SECONDS);
    }

    /**
     * 根据key获取map
     *
     * @param key the key
     */
    @Override
    public Map getMap(String key) {
        return getMap(key, NOT_EXPIRE);
    }

    /**
     * 根据key获取map
     *
     * @param key  the key
     * @param time 过期时间(秒)
     */
    @Override
    public Map getMap(String key, long time) {
        return getMap(key, time, TimeUnit.SECONDS);
    }

    /**
     * 根据key获取map
     *
     * @param key      the key
     * @param time     过期时间
     * @param timeUnit 过期时间单位
     */
    @Override
    public Map getMap(String key, long time, TimeUnit timeUnit) {
        Map result = hashOperations.entries(key);
        if (result.size() == 0) {
            time = 0;
            result = null;
        }
        if (NOT_EXPIRE != time) {
            redisTemplate.expire(key, time, timeUnit);
        }
        return result;
    }

    /**
     * 添加Map
     *
     * @param key     the key
     * @param value   the value
     */
    @Override
    public void setMap(String key, Map value) {
        setMap(key, value, NOT_EXPIRE);
    }

    /**
     * 添加Map
     *
     * @param key     the key
     * @param value   the value
     */
    @Override
    public void setMap(String key, Map value, long time) {
        setMap(key, value, time, TimeUnit.SECONDS);
    }

    @Override
    public void setMap(String key, Map value, long time, TimeUnit timeUnit) {
        ZLock zLock = redissonDistributedLock.lock(key, true);
        try {
            hashOperations.putAll(key, value);
            if (NOT_EXPIRE != time) {
                redisTemplate.expire(key, time, timeUnit);
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            redissonDistributedLock.unlock(zLock);
        }
    }

    /**
     * 获取单个hash key对应的值
     *
     * @param key     the key
     * @param hashKey the hash key
     */
    @Override
    public String getHash(String key, String hashKey) {
        return getHash(key, hashKey, String.class);
    }

    /**
     * 获取单个hash key对应的值
     *
     * @param key     the key
     * @param hashKey the hash key
     */
    @Override
    public <T> T getHash(String key, String hashKey, Class<T> clazz) {
        try {
            String value = (String) redisTemplate.boundHashOps(key).get(hashKey);
            return (null == value) ? null : fromJson(value, clazz);
        } catch (Exception e) {
            return null;
        }
    }

    /**
     * 批量添加
     *
     * @param key     the key
     * @param hashKey the hash key
     * @param value   the value
     */
    @Override
    public void setHash(String key, String hashKey, Object value) {
        setHash(key, hashKey, value, NOT_EXPIRE);
    }

    /**
     * 批量添加
     *
     * @param key     the key
     * @param hashKey the hash key
     * @param value   the value
     * @param time    过期时间(秒)
     */
    @Override
    public void setHash(String key, String hashKey, Object value, long time) {
        setHash(key, hashKey, value, time, TimeUnit.SECONDS);
    }

    /**
     * 批量添加
     *
     * @param key      the key
     * @param hashKey  the hash key
     * @param value    the value
     * @param time     过期时间
     * @param timeUnit 过期时间单位
     */
    @Override
    public void setHash(String key, String hashKey, Object value, long time, TimeUnit timeUnit) {
        ZLock zLock = redissonDistributedLock.lock(key, true);
        try {
            redisTemplate.boundHashOps(key).put(hashKey, toJson(value));
            if (NOT_EXPIRE != time) {
                redisTemplate.expire(key, time, timeUnit);
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            redissonDistributedLock.unlock(zLock);
        }
    }

    /**
     * 根据key值删除
     *
     * @param key      the key
     * @param hashKeys the hash keys
     */
    @Override
    public void delHash(String key, Object... hashKeys) {
        ZLock zLock = redissonDistributedLock.lock(key, true);
        try {
            hashOperations.delete(key, hashKeys);
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            redissonDistributedLock.unlock(zLock);
        }
    }

    /**
     * 主键对应的值加一,value值必须是全数字的字符串
     */
    @Override
    public long increment(final String key) {
        ZLock zLock = redissonDistributedLock.lock(key, true);
        try {
            return valueOperations.increment(key);
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            redissonDistributedLock.unlock(zLock);
        }
    }

    /**
     * 根据key存储list
     */
    @Override
    public void pushList(String key, List list) {
        pushList(key, list, NOT_EXPIRE);
    }

    /**
     * 根据key存储list
     */
    @Override
    public void pushList(String key, List list, long time) {
        pushList(key, list, time, TimeUnit.SECONDS);
    }

    /**
     * 根据key存储list
     */
    @Override
    public void pushList(String key, List list, long time, TimeUnit timeUnit) {
        ZLock zLock = redissonDistributedLock.lock(key, true);
        try {
            listOperations.rightPushAll(key, list);
            if (NOT_EXPIRE != time) {
                redisTemplate.expire(key, time, timeUnit);
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            redissonDistributedLock.unlock(zLock);
        }
    }

    /**
     * 将value 插入到列表 key 的表头
     */
    @Override
    public Long leftPush(String key, Object value) {
        ZLock zLock = redissonDistributedLock.lock(key, true);
        try {
            return listOperations.leftPush(key, value);
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            redissonDistributedLock.unlock(zLock);
        }
    }

    /**
     * 移除并返回列表 key 的表头
     */
    @Override
    public Object leftPop(String key) {
        ZLock zLock = redissonDistributedLock.lock(key, true);
        try {
            return listOperations.leftPop(key);
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            redissonDistributedLock.unlock(zLock);
        }
    }

    /**
     * 将value 插入到列表 key 的表尾
     */
    @Override
    public Long rightPush(String key, Object value) {
        ZLock zLock = redissonDistributedLock.lock(key, true);
        try {
            return listOperations.rightPush(key, value);
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            redissonDistributedLock.unlock(zLock);
        }
    }

    /**
     * 移除并返回列表 key 的最后一个元素
     */
    @Override
    public Object rightPop(String key) {
        ZLock zLock = redissonDistributedLock.lock(key, true);
        try {
            return listOperations.rightPop(key);
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            redissonDistributedLock.unlock(zLock);
        }
    }

    /**
     * 返回列表 key 的长度
     * 如果 key 不存在，则返回 0 ; 如果 key 不是列表类型，返回一个错误
     */
    @Override
    public Long listLength(String key) {
        ZLock zLock = redissonDistributedLock.lock(key, true);
        try {
            return listOperations.size(key);
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            redissonDistributedLock.unlock(zLock);
        }
    }

    /**
     * 将列表 key 下标为 index 的元素的值设置为 value
     *
     * @param key   the key
     * @param index the index
     * @param value the value
     */
    @Override
    public void updateList(String key, long index, Object value) {
        ZLock zLock = redissonDistributedLock.lock(key, true);
        try {
            listOperations.set(key, index, value);
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            redissonDistributedLock.unlock(zLock);
        }
    }

    /**
     * 根据参数 count 的值，移除列表中与参数 value 相等的元素
     *
     * @param key   the key
     * @param count count > 0 : 从表头开始向表尾搜索，移除与 value 相等的元素，数量为 count 。
     *              count < 0 : 从表尾开始向表头搜索，移除与 value 相等的元素，数量为 count 的绝对值。
     *              count = 0 : 移除表中所有与 VALUE 相等的值。
     * @param value the value
     */
    @Override
    public void removeList(String key, long count, Object value) {
        ZLock zLock = redissonDistributedLock.lock(key, true);
        try {
            listOperations.remove(key, count, value);
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            redissonDistributedLock.unlock(zLock);
        }
    }

    /**
     * 根据key返回列表
     */
    @Override
    public List getList(String key) {
        return getList(key, 0, -1);
    }

    /**
     * 返回列表 key 中指定区间内的元素，区间以偏移量 start 和 end 指定。
     *
     * @param key   the key
     * @param start the start
     * @param end   the end
     */
    @Override
    public List getList(String key, int start, int end) {
        return listOperations.range(key, start, end);
    }

    /**
     * 返回列表 key 中指定区间内的元素，区间以偏移量 start 和 end 指定。
     *
     * @param key             the key
     * @param valueSerializer 序列化
     */
    public List getList(String key, RedisSerializer<Object> valueSerializer) {
        return getList(key, 0, -1, valueSerializer);
    }

    /**
     * 返回列表 key 中指定区间内的元素，区间以偏移量 start 和 end 指定。
     *
     * @param key             the key
     * @param start           the start
     * @param end             the end
     * @param valueSerializer 序列化
     */
    public List getList(String key, int start, int end, RedisSerializer<Object> valueSerializer) {
        byte[] rawKey = rawKey(key);
        return (List) redisTemplate.execute(connection -> deserializeValues(connection.lRange(rawKey, start, end), valueSerializer), true);
    }

    /**
     * 查询集合数量
     */
    @Override
    public long dbSize() {
        return (Long) redisTemplate.execute(RedisServerCommands::dbSize);
    }

    /**
     * 判断某个主键是否存在
     */
    @Override
    public boolean exists(final String key) {
        return redisTemplate.hasKey(key);
    }

    /**
     * 删除key
     */
    @Override
    public boolean delete(final String... keys) {
        ZLock zLock = redissonDistributedLock.lock(Convert.toString(keys), true);
        try {
            boolean result = false;
            for (String key : keys) {
                result = redisTemplate.delete(key);
            }
            return result;
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            redissonDistributedLock.unlock(zLock);
        }
    }

    /**
     * 清空redis存储的数据
     */
    @Override
    public void flushDB() {
        redisTemplate.execute((RedisCallback<String>) connection -> {
            connection.flushDb();
            return null;
        });
    }

    /**
     * 清空DB
     */
    public void flushDB(RedisClusterNode node) {
        this.redisTemplate.opsForCluster().flushDb(node);
    }

    private byte[] rawKey(Object key) {
        Assert.notNull(key, "non null key required");

        if (key instanceof byte[]) {
            return (byte[]) key;
        }
        RedisSerializer<Object> redisSerializer = (RedisSerializer<Object>) redisTemplate.getKeySerializer();
        return redisSerializer.serialize(key);
    }

    private byte[] rawValue(Object value, RedisSerializer valueSerializer) {
        if (value instanceof byte[]) {
            return (byte[]) value;
        }

        return valueSerializer.serialize(value);
    }

    private List deserializeValues(List<byte[]> rawValues, RedisSerializer<Object> valueSerializer) {
        if (null == valueSerializer) {
            return rawValues;
        }
        return SerializationUtils.deserialize(rawValues, valueSerializer);
    }

    private Object deserializeValue(byte[] value, RedisSerializer<Object> valueSerializer) {
        if (null == valueSerializer) {
            return value;
        }
        return valueSerializer.deserialize(value);
    }

    /**
     * Object转成JSON数据
     */
    private String toJson(Object object) {
        if (object instanceof Integer || object instanceof Long || object instanceof Float ||
                object instanceof Double || object instanceof Boolean || object instanceof String) {
            return String.valueOf(object);
        }

        return JsonUtil.toJson(object);
    }

    /**
     * JSON数据，转成Object
     */
    private <T> T fromJson(String json, Class<T> clazz) {
        return JsonUtil.toBean(json, clazz);
    }
}
